package com.gengzp.mq.server.service.impl;

import com.gengzp.mq.server.enums.MqMessageStatusEnums;
import com.gengzp.mq.server.facade.MqTopicFacadeService;
import com.gengzp.mq.server.manager.MqAckMessageManager;
import com.gengzp.mq.server.manager.MqBlockWaitingManager;
import com.gengzp.mq.server.mapper.MqMessageMapper;
import com.gengzp.mq.server.model.generic.MqPaginationResult;
import com.gengzp.mq.server.model.message.entity.MqMessage;
import com.gengzp.mq.server.model.message.req.*;
import com.gengzp.mq.server.model.message.resp.MqPageMessageResp;
import com.gengzp.mq.server.model.message.resp.MqReceiveMessageResp;
import com.gengzp.mq.server.model.topic.entity.MqTopic;
import com.gengzp.mq.server.service.MqMessageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;

import static com.gengzp.mq.server.constants.MqWaitTimeConstants.INITIAL_RESEND_INTERVAL;
import static com.gengzp.mq.server.constants.MqWaitTimeConstants.WAIT_TIME;

/**
 * @ClassName MqMessageServiceImpl
 * @Description 消息管理服务层实现类
 * @Author gengzp
 * @Date 2025/9/4 15:28
 */
@Service
public class MqMessageServiceImpl implements MqMessageService {

    private static final Logger logger = LoggerFactory.getLogger(MqMessageServiceImpl.class);

    @Autowired
    private MqMessageMapper mqMessageMapper;

    @Autowired
    private MqBlockWaitingManager mqBlockWaitingManager;

    @Autowired
    private MqAckMessageManager mqAckMessageManager;

    @Autowired
    private MqTopicFacadeService mqTopicFacadeService;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public String sendMessage(MqSendMessageReq req) {
        // 1.查询主题是否存在, 若不存在则创建默认主题
        MqTopic mqTopic = mqTopicFacadeService.searchTopicByName(req.getTopic());
        if (Objects.isNull(mqTopic)) {
            mqTopicFacadeService.autoCreateTopic(req.getTopic());
        }

        // 2.生成消息唯一标识
        String messageId = UUID.randomUUID().toString().replace("-", "");

        // 3.组装消息对象
        MqMessage mqMessage = new MqMessage();
        mqMessage.setMessageId(messageId);
        mqMessage.setTopic(req.getTopic());
        mqMessage.setTag(req.getTag());
        mqMessage.setContent(req.getContent());
        mqMessage.setStatus(MqMessageStatusEnums.UNCONSUMED.getCode());
        mqMessage.setSendTimeStamp(new Date().getTime());
        mqMessage.setRetryCount(0);
        mqMessage.setNextResendIntervalTime(INITIAL_RESEND_INTERVAL);

        // 4.保存消息
        mqMessageMapper.insertMqMessage(mqMessage);

        // 5.唤醒正在等待的消费者线程
        mqBlockWaitingManager.notifyMessageWaiter(mqMessage);

        logger.info("生产一条消息, 消息id: " + messageId);
        return messageId;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public MqReceiveMessageResp receiveMessage(MqReceiveMessageReq req) throws Exception {
        // 1.根据 topic 获取最旧的存量消息
        MqMessage mqMessage = mqMessageMapper.selectOldestUnConsumedMessageByTopic(req.getTopic());

        // 2.若没有存量消息, 那么调用等待管理器进行阻塞等待, 超时则返回null
        if (mqMessage == null) {
            mqMessage = mqBlockWaitingManager.waitForMessage(req.getTopic(), WAIT_TIME);
        }

        // 3.若无消息, 则返回空值
        if (mqMessage == null) {
            return null;
        }

        // 4.获取到消息, 则更改消息状态为消费中
        mqMessageMapper.updateMessageStatusById(
                mqMessage.getMessageId(), MqMessageStatusEnums.CONSUMING.getCode());
        MqReceiveMessageResp resp = new MqReceiveMessageResp();
        resp.setMessageId(mqMessage.getMessageId());
        resp.setTag(mqMessage.getTag());
        resp.setContent(mqMessage.getContent());
        logger.info("消费一条消息, 消息id: " + mqMessage.getMessageId());

        // 5.发送消息到ack超时检查队列
        mqAckMessageManager.sendMessageToAckTimeoutInspectJobQueue(mqMessage);
        return resp;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void ackMessage(MqAckMessageReq req) {
        mqMessageMapper.updateMessageStatusById(
                req.getMessageId(), MqMessageStatusEnums.CONSUME_SUCCESS.getCode());
        logger.info("确认一条消息, 消息id: " + req.getMessageId());
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void resendMessage(MqResendMessageReq req) {
        MqMessage mqMessage = mqMessageMapper.selectMqMessageById(req.getMessageId());
        if (mqMessage == null) {
            throw new RuntimeException("消息id不存在");
        }
        if (mqMessage.getStatus().equals(MqMessageStatusEnums.CONSUMING.getCode())) {
            throw new RuntimeException("消息正在消费中, 无法重复操作");
        }

        // 校验主题是否存在, 若不存在则自动创建
        if (Objects.isNull(mqTopicFacadeService.searchTopicByName(mqMessage.getTopic()))) {
            mqTopicFacadeService.autoCreateTopic(mqMessage.getTopic());
        }

        mqMessage.setRetryCount(mqMessage.getRetryCount() + 1);
        mqMessage.setStatus(MqMessageStatusEnums.UNCONSUMED.getCode());
        mqMessageMapper.updateMessageById(mqMessage);
        logger.info("重发一条消息, 消息id: " + req.getMessageId());
    }

    @Override
    public MqPaginationResult<MqPageMessageResp> pageQueryMessage(MqPageMessageReq req) {
        List<MqPageMessageResp> result = new ArrayList<>();

        // 计算偏移量
        req.setOffset((req.getPageNum() - 1) * req.getPageSize());

        List<MqMessage> mqMessages = mqMessageMapper.pageQueryMessage(req);
        if (mqMessages != null && !mqMessages.isEmpty()) {
            result = mqMessages.stream().map(v -> {
                MqPageMessageResp resp = new MqPageMessageResp();
                resp.setMessageId(v.getMessageId());
                resp.setTopic(v.getTopic());
                resp.setTag(v.getTag());
                resp.setContent(v.getContent());
                resp.setStatus(v.getStatus());
                Optional.ofNullable(MqMessageStatusEnums.getByCode(v.getStatus()))
                        .ifPresent(status -> resp.setStatusDesc(status.getDesc()));
                resp.setSendTime(new Date(v.getSendTimeStamp()));
                resp.setSendTimeStamp(v.getSendTimeStamp());
                resp.setRetryCount(v.getRetryCount());
                return resp;
            }).toList();
        }

        return MqPaginationResult.get(req.getPageNum(), req.getPageSize(), mqMessageMapper.pageQueryMessageCount(req), result);
    }

}
